package com.gjxblog.core;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 保障可靠性投递
 * 确认机制
 */
public class ConfirmTests {
    //2 创建连接
   static Connection connection;

    @Before
    public void before() throws IOException, TimeoutException {
        //1 准备连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("mysql.db.gjxblog.cn");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        //2 创建连接
        connection = connectionFactory.newConnection();

        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.#";
        String queueName = "test_confirm_queue";
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(exchangeName,"topic",true);
        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueBind(queueName,exchangeName,routingKey);
    }



    /**
     * confirm机制
     */
    @Test
    public void testConfirm() throws IOException {

    }

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("mysql.db.gjxblog.cn");
        connectionFactory.setPort(5673);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        //2 创建连接
        Connection  connection = connectionFactory.newConnection();

        Channel channel = connection.createChannel();

        //1 指定我们的消息模式 消息的确认模式
        channel.confirmSelect();

        String exchangeName = "test_confirm_exchange";
        String routingKey = "confirm.save";
        String msg = "Hello RabbitMQ!";

        channel.basicPublish(exchangeName,routingKey,null,msg.getBytes());

        //2 指定监听器
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("ack deliveryTag is：" + deliveryTag);
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("no ack deliveryTag is："+deliveryTag);
            }
        });
    }

    static class Consumer{
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("mysql.db.gjxblog.cn");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            //2 创建连接
           Connection  connection = connectionFactory.newConnection();

            Channel channel = connection.createChannel();


            MyConsumer myConsumer = new MyConsumer(channel);
            channel.basicConsume("test_confirm_queue",true,myConsumer);

            while (true) {

            }
        }
    }

   static class MyConsumer extends DefaultConsumer {
        public MyConsumer(Channel channel){
            super(channel);
        }
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.err.println("-----------consume message----------");
            System.err.println("consumerTag: " + consumerTag);
            System.err.println("envelope: " + envelope);
            System.err.println("properties: " + properties);
            System.err.println("body: " + new String(body));
        }
    }
}
